package com.atguigu.tingshu.search.service.impl;

import co.elastic.clients.elasticsearch.ElasticsearchClient;
import co.elastic.clients.elasticsearch._types.FieldValue;
import co.elastic.clients.elasticsearch._types.SortOrder;
import co.elastic.clients.elasticsearch._types.aggregations.Aggregate;
import co.elastic.clients.elasticsearch._types.query_dsl.BoolQuery;
import co.elastic.clients.elasticsearch.core.SearchRequest;
import co.elastic.clients.elasticsearch.core.SearchResponse;
import co.elastic.clients.elasticsearch.core.search.Hit;
import co.elastic.clients.elasticsearch.core.search.Suggestion;
import co.elastic.clients.json.JsonData;
import com.alibaba.fastjson.JSON;
import com.atguigu.tingshu.album.client.AlbumInfoFeignClient;
import com.atguigu.tingshu.album.client.CategoryFeignClient;
import com.atguigu.tingshu.common.config.redis.RedisConfig;
import com.atguigu.tingshu.common.constant.RedisConstant;
import com.atguigu.tingshu.common.result.Result;
import com.atguigu.tingshu.model.album.*;
import com.atguigu.tingshu.model.search.AlbumInfoIndex;
import com.atguigu.tingshu.model.search.AttributeValueIndex;
import com.atguigu.tingshu.model.search.SuggestIndex;
import com.atguigu.tingshu.query.search.AlbumIndexQuery;
import com.atguigu.tingshu.search.repository.AlbumIndexRepository;
import com.atguigu.tingshu.search.repository.SuggestIndexRepository;
import com.atguigu.tingshu.search.service.SearchService;
import com.atguigu.tingshu.user.client.UserInfoFeignClient;
import com.atguigu.tingshu.vo.album.AlbumStatVo;
import com.atguigu.tingshu.vo.search.AlbumInfoIndexVo;
import com.atguigu.tingshu.vo.search.AlbumSearchResponseVo;
import com.atguigu.tingshu.vo.user.UserInfoVo;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.redis.core.RedisTemplate;
import org.springframework.stereotype.Service;
import org.springframework.util.Assert;
import org.springframework.util.CollectionUtils;
import org.springframework.util.StringUtils;

import java.io.IOException;
import java.util.*;
import java.util.concurrent.CompletableFuture;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.stream.Collectors;


@Slf4j
@Service
@SuppressWarnings({"all"})
public class SearchServiceImpl implements SearchService {
    @Autowired
    private AlbumInfoFeignClient albumInfoFeignClient;

    @Autowired
    private UserInfoFeignClient userInfoFeignClient;

    @Autowired
    private CategoryFeignClient categoryFeignClient;

    @Autowired
    private AlbumIndexRepository albumIndexRepository;

    @Autowired
    private ThreadPoolExecutor threadPoolExecutor;

    @Autowired
    private ElasticsearchClient elasticsearchClient;

    @Autowired
    private SuggestIndexRepository suggestIndexRepository;

    @Autowired
    private RedisTemplate redisTemplate;

    @Override
    public List<AlbumInfoIndex> findRankingList(Long category1Id, String dimension) {
        List<AlbumInfoIndex> albumInfoIndexList = (List<AlbumInfoIndex>) redisTemplate.opsForHash().get(RedisConstant.RANKING_KEY_PREFIX + category1Id, dimension);
        return albumInfoIndexList;
    }

    @Override
    public void updateLatelyAlbumRanking() {
        //获取所有一级分类id集合
        Result<List<BaseCategory1>> baseCategory1Result = categoryFeignClient.findAllCategory1();
        Assert.notNull(baseCategory1Result,"远程调用失败");
        List<BaseCategory1> baseCategory1List = baseCategory1Result.getData();
        Assert.notEmpty(baseCategory1List,"查询失败");
        for (BaseCategory1 baseCategory1 : baseCategory1List) {
           //定义五个维度;将维度变为一个数组
            String[] rankingDimensionArray = new String[]{"hotScore", "playStatNum", "subscribeStatNum", "buyStatNum", "commentStatNum"};
            for (String rankingDimension : rankingDimensionArray) {
                //编写dsl语句
                SearchResponse<AlbumInfoIndex> searchResponse = null;
                try {
                    searchResponse = elasticsearchClient.search(s->s.index("albuminfo")
                                    .query(q->q.term(t->t.field("category1Id").value(baseCategory1.getId())))
                                    .sort(st->st.field(f->f.field(rankingDimension).order(SortOrder.Desc)))
                            ,AlbumInfoIndex.class);
                }catch (IOException e){
                   throw new RuntimeException(e);
                }
            //将数据存储到redis中
                List<AlbumInfoIndex> albumInfoIndexList = searchResponse.hits().hits().stream().map(Hit::source).collect(Collectors.toList());
                //存储数据
                redisTemplate.opsForHash().put(RedisConstant.RANKING_KEY_PREFIX + baseCategory1.getId(), rankingDimension, albumInfoIndexList);
            }

        }
    }

    @Override
    public List<String> completeSuggest(String keyword) {
        //  1.  通过动态生成dsl语句获取到检索结果集     2.  从结果集中获取到title字段数据
        SearchResponse<SuggestIndex> searchResponse =null;
        try {
            searchResponse =elasticsearchClient.search(s->s.index("suggestinfo")
                    .suggest(s1->s1.suggesters("suggestionKeyword",
                            s2->s2.prefix(keyword)
                                    .completion(c->c.field("keyword")
                                            .skipDuplicates(true)
                                            .fuzzy(f->f.fuzziness("auto"))))
                            .suggesters("suggestionkeywordPinyin",s2->s2.prefix(keyword)
                                    .completion(c->c.field("keywordPinyin")
                                            .skipDuplicates(true)
                                            .fuzzy(f->f.fuzziness("auto"))))
                            .suggesters("suggestionkeywordSequence",s2->s2.prefix(keyword)
                                    .completion(c->c.field("keywordSequence")
                                            .skipDuplicates(true)
                                            .fuzzy(f->f.fuzziness("auto"))))
                        )
                    ,SuggestIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //创建集合对象
        ArrayList<String> list = new ArrayList<>();
        //获取提词数据
        list.addAll(getSuggestResultData(searchResponse,"suggestionKeyword"));
        list.addAll(getSuggestResultData(searchResponse,"suggestionkeywordPinyin"));
        list.addAll(getSuggestResultData(searchResponse,"suggestionkeywordSequence"));

        return list;
    }

    private Collection<String> getSuggestResultData(SearchResponse<SuggestIndex> searchResponse, String keyword) {

        //获取到suggest集合列表
        List<Suggestion<SuggestIndex>> suggestionList = searchResponse.suggest().get(keyword);
        if (!CollectionUtils.isEmpty(suggestionList)){
            for (Suggestion<SuggestIndex> suggestIndexSuggestion : suggestionList) {

                List<String> stringList = suggestIndexSuggestion.completion().options().stream().map(option -> {
                    return option.source().getTitle();
                }).collect(Collectors.toList());
                //返回集合数据
                return stringList;
            }
        }
        return null;
    }

    @Override
    public List<Map<String, Object>> channel(Long category1Id) {
        //远程调用获取一级分类下对应的三级分类ID集合
        Result<List<BaseCategory3>> result = categoryFeignClient.findTopBaseCategory3(category1Id);
        Assert.notNull(result, "远程调用失败");
        List<BaseCategory3> category3List = result.getData();
        Assert.notNull(category3List, "查询失败");
        Map<Long, BaseCategory3> category3Map = category3List.stream().collect(Collectors.toMap(baseCategory3 -> baseCategory3.getId(), baseCategory3 -> baseCategory3));
        List<FieldValue> fieldValueList = category3List.stream().map(category3 -> FieldValue.of(category3.getId())).collect(Collectors.toList());
        //编写dsl语句
        SearchResponse<AlbumInfoIndex> response = null;

        try {
            response = elasticsearchClient.search(f-> f.index("albuminfo")
                    .query(q -> q.terms(t-> t.field("category3Id").terms(ts ->ts.value(fieldValueList))))
                    .aggregations("agg_category3Id",a ->a.terms(t->t.field("category3Id"))
                            .aggregations("agg_top6",ag ->ag.topHits(th ->th.size(6)
                                    .sort(s->s.field(f1->f1.field("hotScore").order(SortOrder.Desc)))))), AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //创建一个List<Map>对象
        Aggregate aggCategory3 = response.aggregations().get("agg_category3Id");
        //获取数据
        List<Map<String, Object>> mapList = aggCategory3.lterms().buckets().array().stream().map(termsBucket ->{
            //创建一个Map对象
            HashMap<String, Object> map = new HashMap<>();
            long category3Id = termsBucket.key();
            //获取专辑集合列表
            Aggregate aggTop6 = termsBucket.aggregations().get("agg_top6");
            List<AlbumInfoIndex> albumInfoIndexList = aggTop6.topHits().hits().hits().stream().map(hit -> {
                //获取数据
                JsonData source = hit.source();
                //将数据转换为AlbumInfoIndex对象
                AlbumInfoIndex albumInfoIndex = JSON.parseObject(source.toString(), AlbumInfoIndex.class);
                return albumInfoIndex;
            }).collect(Collectors.toList());
            //baseCategory3表示当前的三级分类对象
            map.put("baseCategory3",category3Map.get(category3Id));
            map.put("list",albumInfoIndexList);
            return map;
        }).collect(Collectors.toList());

        return mapList;
    }

    @Override
    public AlbumSearchResponseVo search(AlbumIndexQuery albumIndexQuery) {
        //1.生成动态DSL语句  2.  根据dsl语句进行查询数据   3.  获取到结果并赋值给AlbumSearchResponseVo对象
        SearchRequest searchRequest= quertBuildDsl(albumIndexQuery);
        SearchResponse<AlbumInfoIndex> searchResponse = null;
        try {
            searchResponse = elasticsearchClient.search(searchRequest, AlbumInfoIndex.class);
        } catch (IOException e) {
            throw new RuntimeException(e);
        }
        //创建AlbumSearchResponseVo对象并赋值
        AlbumSearchResponseVo albumSearchResponseVo = parseResultData(searchResponse);
        albumSearchResponseVo.setTotal(searchResponse.hits().total().value());
        albumSearchResponseVo.setPageSize(albumIndexQuery.getPageSize());
        albumSearchResponseVo.setPageNo(albumIndexQuery.getPageNo());
        //计算总页数
        Long totalPages = (albumSearchResponseVo.getTotal() + albumIndexQuery.getPageSize() - 1) / albumIndexQuery.getPageSize();
        albumSearchResponseVo.setTotalPages(totalPages);
        //返回对象
        return albumSearchResponseVo;
    }


    /**
     *生成动态的DSL语句
     */
    private SearchRequest quertBuildDsl(AlbumIndexQuery albumIndexQuery) {
        //创建一个searchRequest对象
        SearchRequest.Builder searchRequestBuilder = new SearchRequest.Builder();
        searchRequestBuilder.index("albuminfo");
        //创建一个BoolQuery对象
        BoolQuery.Builder boolQueryBuilder = new BoolQuery.Builder();
        //判断是否通过关键词检索
        String keyword = albumIndexQuery.getKeyword();
        if (!StringUtils.isEmpty(keyword)){
            boolQueryBuilder.should(s->s.match(m->m.field("albumTitle").query(keyword)))
                    .should(s->s.match(m->m.field("albumIntro").query(keyword)))
                    .should(s->s.match(m->m.field("announcerName").query(keyword)));
            //高亮
            searchRequestBuilder.highlight(h->h.fields("albumTitle",h1->h1.preTags("<span style=color:red>").postTags("</span>")));
        }
        //判断是否根据分类id进行过滤
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory1Id())){
            boolQueryBuilder.filter(f->f.term(t->t.field("category1Id").value(albumIndexQuery.getCategory1Id())));
        }
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory2Id())){
            boolQueryBuilder.filter(f->f.term(t->t.field("category2Id").value(albumIndexQuery.getCategory2Id())));
        }
        if (!StringUtils.isEmpty(albumIndexQuery.getCategory3Id())){
            boolQueryBuilder.filter(f->f.term(t->t.field("category3Id").value(albumIndexQuery.getCategory3Id())));
        }
        //根据属性与属性值进行过滤! 属性id:属性值id,使用字符串的split()进行分割
        List<String> attributeList = albumIndexQuery.getAttributeList();
        if (!CollectionUtils.isEmpty(attributeList)){
            for (String attr : attributeList) {
                //attr进行分割
                String[] split = attr.split(":");
                //判断
                if (null != split && split.length == 2){
                    //属性Id
                    String attrId = split[0];
                    //属性值Id
                    String attrValueId = split[1];
                    boolQueryBuilder.filter(f->f.nested(n->n.path("albumAttributeValueList")
                            .query(qy->qy.bool(b->b.filter(f1->f1.term(t->t.field("attributeValueIndexList.attributeId").value(attrId)))
                                    .filter(f2->f2.term(t->t.field("attributeValueIndexList.valueId").value(attrValueId)))))));
                }

            }
        }

        //设置排序: 综合排序[1:desc] 播放量[2:desc] 发布时间[3:desc] ; asc:升序 desc:降序
        String order = albumIndexQuery.getOrder();
        if (!StringUtils.isEmpty(order)){
            //分割
            String[] split = order.split(":");
            if (null != split && split.length == 2){
                //声明一个字段
                String field = "";
                switch (split[0]){
                    case "1":
                        field = "hotScore";
                        break;
                    case "2":
                        field = "playStatNum";
                        break;
                    case "3":
                        field = "creatTime";
                        break;
                }
                //重新声明变量
                String finalField = field;
                searchRequestBuilder.sort(s->s.field(f->f.field(finalField).order("asc".equals(split[1])? SortOrder.Asc: SortOrder.Desc)));
            }
        }

        //分页
        searchRequestBuilder.from((albumIndexQuery.getPageNo() - 1) * albumIndexQuery.getPageSize());
        searchRequestBuilder.size(albumIndexQuery.getPageSize());
        BoolQuery boolQuery=boolQueryBuilder.build();
        //调用query()
        searchRequestBuilder.query(boolQuery._toQuery());
        //调用build();方法创建对象
        SearchRequest searchRequest = searchRequestBuilder.build();
        System.out.println("dsl:"+searchRequest.toString());
        //  返回对象
        return searchRequest;
    }

    /**
     * 赋值专辑集合数据
     * @param searchResponse
     * @return
     */
    private AlbumSearchResponseVo parseResultData(SearchResponse<AlbumInfoIndex> searchResponse) {
        //创建对象
        AlbumSearchResponseVo albumSearchResponseVo = new AlbumSearchResponseVo();
        //创建一个集合
        ArrayList<AlbumInfoIndexVo> list = new ArrayList<>();
        for (Hit<AlbumInfoIndex> hit : searchResponse.hits().hits()) {
            //创建一个AlbumInfoIndexVo对象
            AlbumInfoIndexVo albumInfoIndexVo = new AlbumInfoIndexVo();
            //获取索引库的数据
            AlbumInfoIndex albumInfoIndex = hit.source();
            //属性拷贝
            BeanUtils.copyProperties(albumInfoIndex, albumInfoIndexVo);
            //判断是否高亮
            if (null != hit.highlight().get("albumTitle")){
                //获取高亮字段
                String albumTitle = hit.highlight().get("albumTitle").get(0);
                //赋值
                albumInfoIndexVo.setAlbumTitle(albumTitle);
            }
            list.add(albumInfoIndexVo);
        }
        //添加到集合中
        albumSearchResponseVo.setList(list);
        //返回
        return albumSearchResponseVo;



    }

    @Override
    public void upperAlbum(Long albumId) {

        //给AlbumInfoIndex赋值
        AlbumInfoIndex albumInfoIndex = new AlbumInfoIndex();
        CompletableFuture<AlbumInfo> albumInfoCompletableFuture = CompletableFuture.supplyAsync(() -> {
            //远程调用专辑微服务获取专辑对象
            Result<AlbumInfo> albumInfoResult = albumInfoFeignClient.getAlbumInfo(albumId);
            //判断专辑对象是否为空
            Assert.notNull(albumInfoResult, "专辑结果集不存在");
            AlbumInfo albumInfo = albumInfoResult.getData();
            Assert.notNull(albumInfo, "专辑对象不存在");
            //属性拷贝
            BeanUtils.copyProperties(albumInfo, albumInfoIndex);
            return albumInfo;
        }, threadPoolExecutor);


        //获取用户信息
        CompletableFuture<Void> userCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //远程调用
            Result<UserInfoVo> userInfoVoResult = userInfoFeignClient.getUserInfoVo(albumInfo.getUserId());
            //判断
            Assert.notNull(userInfoVoResult, "用户信息不存在");
            UserInfoVo userInfoVo = userInfoVoResult.getData();
            Assert.notNull(userInfoVo, "用户信息对象不存在");
            albumInfoIndex.setAnnouncerName(userInfoVo.getNickname());
        }, threadPoolExecutor);


        //获取分类数据
        CompletableFuture<Void> cateCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            Result<BaseCategoryView> baseCategoryViewResult = categoryFeignClient.getCategoryView(albumInfo.getCategory3Id());
            Assert.notNull(baseCategoryViewResult, "分类数据不存在");
            BaseCategoryView baseCategoryView = baseCategoryViewResult.getData();
            Assert.notNull(baseCategoryView, "分类对象不存在");
            albumInfoIndex.setCategory1Id(baseCategoryView.getCategory1Id());
            albumInfoIndex.setCategory2Id(baseCategoryView.getCategory2Id());
        }, threadPoolExecutor);

        //播放量
        CompletableFuture<Void> statCompletableFuture = albumInfoCompletableFuture.runAsync(() -> {
            Result<AlbumStatVo> albumStatistics = albumInfoFeignClient.getAlbumStatistics(albumId);
            AlbumStatVo albumStatVo = albumStatistics.getData();
            albumInfoIndex.setPlayStatNum(albumStatVo.getPlayStatNum());
            albumInfoIndex.setCommentStatNum(albumStatVo.getCommentStatNum());
            albumInfoIndex.setSubscribeStatNum(albumStatVo.getSubscribeStatNum());
            albumInfoIndex.setBuyStatNum(albumStatVo.getBuyStatNum());
            //热度
            albumInfoIndex.setHotScore(new Random().nextDouble() * 100);

        }, threadPoolExecutor);

        //获取属性数据
        CompletableFuture<Void> attrCompletableFuture = albumInfoCompletableFuture.thenAcceptAsync(albumInfo -> {
            //获取专辑对应的属性与属性值集合

            List<AlbumAttributeValue> albumAttributeValueVoList = albumInfo.getAlbumAttributeValueVoList();
            //遍历
            if (!CollectionUtils.isEmpty(albumAttributeValueVoList)) {
                List<AttributeValueIndex> attributeValueIndexList = albumAttributeValueVoList.stream().map(albumAttributeValue -> {
                    //  创建对象
                    AttributeValueIndex attributeValueIndex = new AttributeValueIndex();
                    //  属性拷贝
                    BeanUtils.copyProperties(albumAttributeValue, attributeValueIndex);
                    //  返回数据
                    return attributeValueIndex;
                }).collect(Collectors.toList());
                //  赋值：
                albumInfoIndex.setAttributeValueIndexList(attributeValueIndexList);
            }
        }, threadPoolExecutor);


        //数据汇总
        CompletableFuture.allOf(
            albumInfoCompletableFuture,
            userCompletableFuture,
            cateCompletableFuture,
            attrCompletableFuture,
            statCompletableFuture
        ).join();
        //保存到es
        albumIndexRepository.save(albumInfoIndex);
    }

    @Override
    public void lowerAlbum(Long albumId) {
        albumIndexRepository.deleteById(albumId);
    }

    }


